#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
@Project ：python_learning 
@File ：多进程演示demo.py
@IDE  ：PyCharm 
@Author ：李涵彬
@Date ：2025/2/24 下午2:36 
"""

import threading
import time
import random
from queue import Queue
import concurrent.futures
import logging

# 配置日志格式，方便观察线程执行顺序
logging.basicConfig(
	level=logging.INFO,
	format='%(asctime)s [%(threadName)s] %(message)s'
)


# 示例1：基本线程创建与启动
def basic_thread_demo():
	def worker():
		logging.info("开始执行任务")
		time.sleep(1)
		logging.info("任务执行完成")

	# 创建线程
	threads = []
	for i in range(3):
		t = threading.Thread(target=worker, name=f"Worker-{i + 1}")
		threads.append(t)
		t.start()

	# 等待所有线程完成
	for t in threads:
		t.join()


# 示例2：带参数的线程和锁的使用
def lock_demo():
	shared_counter = 0
	lock = threading.Lock()

	def increment_counter(name):
		nonlocal shared_counter
		logging.info(f"{name} 尝试获取锁")
		with lock:  # 自动获取和释放锁
			logging.info(f"{name} 获得锁")
			local_copy = shared_counter
			time.sleep(0.1 * random.random())
			local_copy += 1
			shared_counter = local_copy
			logging.info(f"{name} 更新计数器为: {shared_counter}")

	threads = []
	for i in range(5):
		t = threading.Thread(
			target=increment_counter,
			args=(f"Thread-{i + 1}",)
		)
		threads.append(t)
		t.start()

	for t in threads:
		t.join()
	print(f"最终计数器值: {shared_counter}")


# 示例3：信号量（Semaphore）使用
def semaphore_demo():
	sem = threading.Semaphore(2)  # 同时允许2个线程访问

	def limited_resource(user):
		with sem:
			logging.info(f"{user} 获得资源访问")
			time.sleep(1 + random.random())
			logging.info(f"{user} 释放资源访问")

	users = [f"User-{i}" for i in range(5)]
	threads = []
	for user in users:
		t = threading.Thread(target=limited_resource, args=(user,))
		threads.append(t)
		t.start()

	for t in threads:
		t.join()


# 示例4：事件（Event）机制
def event_demo():
	event = threading.Event()

	def waiter():
		logging.info("等待事件触发...")
		event.wait()
		logging.info("事件已触发，继续执行")

	def setter():
		time.sleep(2)
		logging.info("事件触发！")
		event.set()

	t1 = threading.Thread(target=waiter, name="Waiter")
	t2 = threading.Thread(target=setter, name="Setter")
	t1.start()
	t2.start()
	t1.join()
	t2.join()


# 示例5：线程局部数据
def thread_local_demo():
	local_data = threading.local()

	def show_data():
		try:
			val = local_data.value
		except AttributeError:
			logging.info("没有本地数据")
		else:
			logging.info(f"本地数据值: {val}")

	def worker(value):
		show_data()
		local_data.value = value
		show_data()

	threads = []
	for i in range(3):
		t = threading.Thread(target=worker, args=(i,), name=f"Local-{i}")
		threads.append(t)
		t.start()

	for t in threads:
		t.join()


# 示例6：队列（Queue）线程通信
def queue_demo():
	q = Queue(maxsize=3)

	def producer():
		for item in ['A', 'B', 'C', 'D', 'E']:
			q.put(item)
			logging.info(f"生产: {item}")
			time.sleep(0.1)

	def consumer():
		while True:
			item = q.get()
			if item is None:  # 终止信号
				break
			logging.info(f"消费: {item}")
			q.task_done()

	producer_thread = threading.Thread(target=producer)
	consumer_thread = threading.Thread(target=consumer)

	producer_thread.start()
	consumer_thread.start()

	producer_thread.join()
	q.put(None)  # 发送终止信号
	consumer_thread.join()


# 示例7：线程池执行器
def thread_pool_demo():
	def task(n):
		logging.info(f"处理任务 {n}")
		time.sleep(0.5)
		return n ** 2

	with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
		futures = [executor.submit(task, i) for i in range(6)]
		results = []
		for future in concurrent.futures.as_completed(futures):
			result = future.result()
			results.append(result)
			logging.info(f"得到结果: {result}")

	print("所有任务结果:", results)


if __name__ == "__main__":
	print("=" * 40)
	print("基本线程演示:")
	basic_thread_demo()

	print("\n" + "=" * 40)
	print("锁机制演示:")
	lock_demo()

	print("\n" + "=" * 40)
	print("信号量演示:")
	semaphore_demo()

	print("\n" + "=" * 40)
	print("事件机制演示:")
	event_demo()

	print("\n" + "=" * 40)
	print("线程本地存储演示:")
	thread_local_demo()

	print("\n" + "=" * 40)
	print("队列通信演示:")
	queue_demo()

	print("\n" + "=" * 40)
	print("线程池演示:")
	thread_pool_demo()
